package com.atguigu.transform;

import com.atguigu.bean.WaterSensor;
import com.atguigu.functions.mapFunctionImpl;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class mapDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 11),
                new WaterSensor("s2", 2L, 22),
                new WaterSensor("s3", 3L, 33)
        );

        //todo map算子：一进一出
        // todo 方式一 ：匿名实现类
//        SingleOutputStreamOperator<String> mapDS = sensorDS.map(
//                new MapFunction<WaterSensor, String>() {
//                    @Override
//                    public String map(WaterSensor waterSensor) throws Exception {
//                        return waterSensor.getId();
//                    }
//                }
//        );
//        mapDS.print();

        //todo 方式2：lambda表达式
//        sensorDS.map(sensor -> sensor.getId()).print();

        // todo 方式3 ：定义一个类来实现MapFunction
        sensorDS.map(new mapFunctionImpl()).print();

        env.execute();
    }
}
